package com.alison.sparkstream.transform

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}

object E2_transform2 {

  """
    |有状态转换操作1：UpdateStateByKey
    |
    |1，基本介绍
    |（1）有时我们需要在 DStream 中跨批次维护状态（例如流计算中累加 wordcount）。针对这种情况，updateStateByKey() 为我们提供了对一个状态变量的访问，用于键值对形式的 DStream。
    |给定一个由 (键，事件) 对构成的 DStream，并传递一个指定如何根据新的事件更新每个键对应状态的函数，它可以构建出一个新的 DStream，其内部数据为 (键，状态) 对。
    |（2）updateStateByKey() 的结果会是一个新的 DStream，其内部的 RDD 序列是由每个时间区间对应的 (键，状态) 对组成的。
    |（3）updateStateByKey() 操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，需要做下面两步：
    |定义状态，状态可以是一个任意的数据类型。
    |定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
    |注意：使用 updateStateByKey 需要对检查点目录进行配置，会使用检查点来保存状态
    |
    |
    |link: https://www.hangge.com/blog/cache/category_83_2.html
    |""".stripMargin

  // 定义状态更新函数,参数 newValues 为当前批次单词频度，runningCount 为以往批次单词频度
  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = newValues.sum + runningCount.getOrElse(0)
    Some(newCount)
  }

  def main(args: Array[String]): Unit = {
//    updatestateByKeyer
    mapwithstateer
  }

  // 创建 Spark 运行配置对象
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

  // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
  val ssc = new StreamingContext(sparkConf, Seconds(5))

  def updatestateByKeyer() = {
    // 设置检查点以保存状态
    ssc.checkpoint("./checkpoint")

    // 通过监控端口创建 DStream，读进来的数据为一行行
    val inputDStream: DStream[String] = ssc.socketTextStream("192.168.56.104", 9999)

    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = inputDStream.flatMap(_.split(" "))

    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))

    // 使用updateStateByKey进行状态更新
    val statefulCounts = wordAndOneStreams.updateStateByKey(updateFunction)

    // 打印结果
    statefulCounts.print()

    // 启动 StreamingContext
    ssc.start()

    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  """
    |mapWithState 操作是一种在 DStream 中应用自定义状态更新和映射函数的方法。它可以用于在连续的时间窗口内，结合之前的状态和新的数据来计算和映射数据。这在需要更复杂的状态处理时非常有用。
    |
    |注意：使用 mapWithState 需要对检查点目录进行配置，会使用检查点来保存状态
    |
    |下面代码中我们定义了一个简单的 StateData 类来存储状态信息，同时使用 mapWithState 来计算累计平均值。
    |
    |提示：要使用 mapWithState，我们首先定义了一个 StateSpec，其中包含了状态更新函数 updateStateFunc。
    |在 updateStateFunc 中，我们将新的值与旧的状态信息相结合，计算新的状态并返回映射结果。
    |
    |
    |运行
    |nc -lk 9999
    |程序启动后，我们在该终端中每隔5秒输入一些数据：
    |a,1
    |b,1
    |//--------等待5秒--------
    |a,2
    |//--------等待5秒--------
    |a,3
    |b,2
    |
    |结果
    |(a,1.0)
    |(b,1.0)
    |-------------5s后
    |(a,1.5)  a,(1+2)/2
    |-------------5s后
    |(a,2.0)
    |(b,1.5)
    |
    |""".stripMargin
  def mapwithstateer = {
    // 设置检查点以保存状态
    ssc.checkpoint("./checkpoint")

    // 从监控端口读取数据流
    val inputDStream = ssc.socketTextStream("192.168.56.104", 9999)

    // 将每行数据转换为键值对，其中键是 parts(0)，值是 parts(1) 转换为 Double 类型
    val mappedDStream = inputDStream.map(line => {
      val parts = line.split(",")
      (parts(0), parts(1).toDouble)
    })

    // 设置初始状态
    val initialRDD = ssc.sparkContext.parallelize(Seq(("key", StateData(0, 0.0))))
    val stateSpec = StateSpec.function(mappingFunction = updateStateFunc _).initialState(initialRDD)

    // 使用 mapWithState 进行状态更新和映射计算
    val statefulDStream = mappedDStream.mapWithState(stateSpec)

    // 打印结果
    statefulDStream.print()

    // 启动 StreamingContext
    ssc.start()

    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }


  // 定义状态数据的样例类
  case class StateData(count: Int, sum: Double)

  /**
   * 更新状态的函数，用于在有状态转换中更新状态和计算结果
   *
   * @param key   输入的键
   * @param value 输入的值，可能为空
   * @param state 用于管理状态的 State 对象
   * @return 更新后的结果，包含键和计算得到的值
   */
  def updateStateFunc(key: String, value: Option[Double], state: State[StateData]):
  (String, Double) = {
    val newValue = value.getOrElse(0.0)
    val oldState = state.getOption().getOrElse(StateData(0, 0.0))
    val newState = StateData(oldState.count + 1, oldState.sum + newValue)
    state.update(newState)
    (key, newState.sum / newState.count)
  }
}
